#!/usr/bin/python3
# coding=utf-8

import time
import pandas as pd
from elasticsearch import Elasticsearch
from random import randint, choice
import requests
import sys

sys.path.append("./")
sys.path.append("../")
sys.dont_write_bytecode = True
from utils.logger_helper import logObj
from utils.config_helper import Platform_Vars
from datetime import datetime, timedelta
import json


def sync_cache_pool_upload_usages(env: str):
    # 基于用量上报的处理机制， 每格1分钟，推送一次无效数据将ES的缓存池的记录推送进DB
    api_fat = (
        "http://bill-data-fat.phoenix-t.xforceplus.com/1400743552728313856/bill-data/v1"
    )
    api_sit = (
        "http://bill-data-sit.phoenix-t.xforceplus.com/1408318096905289728/bill-data/v1"
    )
    api = api_fat if env == "fat" else api_sit
    bizSn = time.strftime("%Y%m%d", time.localtime()) + str(randint(10000000, 99999999))
    tenantId = "1111111111111111111"
    tenantCode = "AAA"
    tenantName = "AAA"
    ocrTime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    ocrFinishTime = ocrTime
    ocrChannel = choice(["API", "PLATFORM"])
    sourceCode = "A_2"
    invoiceCode = time.strftime("%Y%m%d", time.localtime()) + str(randint(1000, 9999))
    invoiceNo = int(time.time())
    companyName = "AAAAAAAAAAAAAAAAAA"
    invoiceType = "NORMAL"
    taxNum = 123456789123456789
    req_header = {"Content-Type": "application/json;charset=UTF8"}
    req_body = {
        "bizSn": bizSn,
        "tenantId": tenantId,
        "tenantCode": tenantCode,
        "tenantName": tenantName,
        "companyName": companyName,
        "invoiceCode": invoiceCode,
        "invoiceNo": invoiceNo,
        "invoiceType": invoiceType,
        "ocrTime": ocrTime,
        "ocrFinishTime": ocrFinishTime,
        "ocrChannel": ocrChannel,
        "sourceCode": sourceCode,
        "taxNum": taxNum,
    }
    req_body_json = json.dumps(req_body)
    # print(req_body_json)
    url = api + "/ocr"
    res = requests.post(url, headers=req_header, data=req_body_json)
    if res.status_code == 200:
        print("Sync Usage Upload result = True")
    else:
        print("Sync Usage Upload result = False")
        exit(0)


class ESearchHelper:
    def __init__(self, hosts, port=9200):
        self.hosts = hosts
        self.port = port
        self.es = Elasticsearch(hosts=self.hosts, port=self.port, ignore=[400, 404])

    def create_index(self, target_index):
        """创建Index 并返回创建结果"""
        is_created = False
        if self.es:
            if self.es.indices.exists(index=target_index):
                self.es.delete(index=target_index)

            result = self.es.indices.create(index=target_index, ignore=[400, 404])
            is_created = result["found"]
            logObj.debug("目标Index:{target_index}, 创建结果:{is_created}")

        return is_created

    def search_indics(self, keyword):
        """查询所有的index名字"""
        asterisk_indics = []
        if self.es:
            asterisk_indics = self.es.indices.exists(index=keyword)
            if asterisk_indics:
                for idx in asterisk_indics:
                    print(idx)

        return asterisk_indics

    def search_by_query(self, target_index, target_query):
        """基于各种查询语句返回数据"""
        records_count = 0
        es_df = None
        if self.es:
            if self.es.indices.exists(index=target_index):
                if target_query:
                    result = self.es.search(
                        index=target_index, body=target_query, size=1000
                    )
                    hits_info = result.get("hits")
                    total = hits_info.get("total")
                    records_count = total.get("value")
                    # Logger.warning(f'>>>>一共找到记录数:{records_count}')
                    if records_count > 0:
                        records_list = hits_info.get("hits")
                        for record_info in records_list:
                            logObj.debug(record_info)
                        es_df = pd.DataFrame(records_list)
                        es_df = pd.DataFrame(list(es_df["_source"]))
                        es_df = es_df[
                            [
                                "tenantCode",
                                "taxNum",
                                "billingDate",
                                "eventId",
                                "bizType",
                            ]
                        ]
                        # print(es_df)

                else:
                    print("查询语句为空，请重新确认。")
            else:
                print("目标Index 不存在, 请重新确认。")

        return records_count, es_df

    def update_by_query(self, target_index, target_query):
        """基于查询结果完成数据修改"""
        updated_count = 0
        if self.es:
            if self.es.indices.exists(index=target_index):
                if target_query:
                    result = self.es.update_by_query(
                        index=target_index, body=target_query, conflicts="proceed"
                    )
                    updated_count = result.get("updated")
                    # Logger.debug(f'一共更新记录数:{updated_count}')
                    logObj.debug("一共更新记录数:{}".format(updated_count))
                else:
                    print("查询语句为空，请重新确认！")
            else:
                print("目标Index 不存在, 请重新确认！")

        return updated_count

    def insert_by_data(self, target_index, target_data):
        """插入数据到指定的Index中
        data = {
            "bizSn": "ocr-1634806665-7111341",
            "tenantId": 5776181997559988224,
            "tenantCode": "Engine-Test-Tenant",
            "tenantName": "123123",
            "companyName": "123123123SS1",
            "taxNum": "991000007N16YUAY5C",
            "invoiceCode": 5834214909,
            "invoiceNo": 10008822461,
            "sourceCode": "A_4",
            "invoiceType": "VEHICLE",
            "ocrTime": "2021-10-21 16:57:45",
            "ocrFinishTime": "2021-10-21 16:57:52",
            "ocrChannel": "API_OLD"
        }
        """
        is_inserted = False
        if self.es:
            if self.es.indices.exists(index=target_index):
                if target_data:
                    result = self.es.index(index=target_index, document=target_data)
                    is_inserted = result["status"]
                else:
                    print("目标数据为空，请重新确认！")
            else:
                print("目标Index 不存在, 请重新确认！")

        return is_inserted

    def delete_index(self, target_index):
        """删除指定index"""
        is_deleted = False
        if self.es:
            result = self.es.delete(index=target_index, ignore=[400, 404])
            is_deleted = result["found"]

        return is_deleted

    def delete_by_query(self, target_index, target_query):
        """删除查询到的数据, 默认指定index存在。"""
        deleted_count = 0
        if self.es:
            if self.es.indices.get(index=target_index):
                if target_query:
                    print(target_query)
                    result = self.es.delete_by_query(
                        index=target_index, body=target_query, conflicts="proceed"
                    )
                    deleted_count = result.get("deleted")
                    # Logger.warning(f'一共删除记录数:{deleted_count}')
                    logObj.warning("一共删除记录数:{}".format(deleted_count))
                else:
                    print("查询语句为空，请重新确认！")
            else:
                print("目标Index 不存在, 请重新确认！")

        return deleted_count


def delete_es_info_by_tenant_code(operator, target_index, tenant_code):
    # 删除ES中所有指定租户在目标Index中的记录
    target_query = {
        "query": {
            "bool": {
                "must": [{"match": {"tenantCode.keyword": {"query": tenant_code}}}]
            }
        }
    }

    # 查询是否有记录并删除
    records_count, _ = operator.search_by_query(target_index, target_query)
    if records_count > 0:
        operator.delete_by_query(target_index, target_query)


def update_es_info_by_tenant_code(
    operator, target_index, tenant_code, bill_date, is_uploaded, exp_count
):
    # source = f"ctx._source.billingDate='{bill_date}'"
    source = "ctx._source.billingDate='{}'".format(bill_date)
    search_query = {
        "query": {
            "bool": {
                "must": [{"match": {"tenantCode.keyword": {"query": tenant_code}}}]
            }
        }
    }
    update_query = {
        "script": {"source": source},
        "query": {
            "bool": {
                "must": [{"match": {"tenantCode.keyword": {"query": tenant_code}}}]
            }
        },
    }

    is_updated = False  # 更新完成状态
    is_verified = False  # 验证更新结果状态

    if is_uploaded and exp_count > 0:
        loop = 0
        while loop <= 600:
            searched_count, _ = operator.search_by_query(target_index, search_query)
            # 判断是否上报的用量记录数 >= 预期记录数
            if searched_count >= exp_count:
                # 进行更新操作, 默认只需要更新一次。
                act_updated_count = operator.update_by_query(target_index, update_query)
                if act_updated_count == searched_count:
                    is_updated = True
                    break

            loop += 30
            time.sleep(30)

        loop = 0
        while is_updated and loop <= 300:
            searched_count, es_df = operator.search_by_query(target_index, search_query)
            billing_dates = es_df["billingDate"]
            sum_updated_count = 0
            for item in billing_dates:
                es_billing_date = item[0:10]
                ref_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
                if es_billing_date == ref_date:
                    sum_updated_count += 1

            if sum_updated_count == act_updated_count:
                is_verified = True
                break

            loop += 30
            time.sleep(30)

    return is_updated, is_verified


def search_records_by_tenant_code(
    env,
    tenant_code,
    tax_num: str = None,
    biz_type: str = None,
    channel: str = None,
    start_date: str = None,
    end_date: str = None,
):
    FAT = ["http://at-es7-fat.phoenix-t.xforceplus.com"]
    SIT = ["http://at-es7-sit.phoenix-t.xforceplus.com"]
    hosts = FAT if env.upper() == "FAT" else SIT
    demo = ESearchHelper(hosts, port=80)
    target_index = "billing"

    search_query = {
        "query": {
            "bool": {
                "must": [{"match": {"tenantCode.keyword": {"query": tenant_code}}}]
            }
        },
        "track_total_hits": True,
        "size": 5000,
    }
    return demo.search_by_query(target_index, search_query)


def delete_records_by_tenant_code(env, tenant_code, exp_index=None):
    FAT = ["http://at-es7-fat.phoenix-t.xforceplus.com"]
    SIT = ["http://at-es7-sit.phoenix-t.xforceplus.com"]
    hosts = FAT if env.upper() == "FAT" else SIT
    demo = ESearchHelper(hosts, port=80)
    target_index = "billing"
    delete_search_query = {
        "query": {
            "bool": {
                "must": [{"match": {"tenantCode.keyword": {"query": tenant_code}}}]
            }
        },
        "track_total_hits": True,
        "size": 5000,
    }
    return demo.delete_by_query(target_index, delete_search_query)


def update_records_by_tenant_code(
    env,
    tenant_code,
    exp_count,
    exp_index=None,
    target_billing_date: str = None,
):
    FAT = ["http://at-es7-fat.phoenix-t.xforceplus.com"]
    SIT = ["http://at-es7-sit.phoenix-t.xforceplus.com"]
    hosts = FAT if env.upper() == "FAT" else SIT
    demo = ESearchHelper(hosts, port=80)
    ref_date = datetime.now() - timedelta(days=1)
    sys_month = ref_date.strftime("%Y.%m")
    # target_index = exp_index if exp_index else 'billing-{}'.format(sys_month)
    target_index = exp_index if exp_index else "billing"
    search_date = ref_date.strftime("%Y-%m-%d")
    search_start_timestamp = search_date + "T00:00:01"
    search_end_timestamp = search_date + "T23:59:59"
    exp_date = ref_date - timedelta(hours=24)
    exp_date_str = exp_date.strftime("%Y-%m-%d %H:%M:%S")
    billing_date = target_billing_date if target_billing_date else exp_date_str
    source = "ctx._source.billingDate='{}'".format(billing_date)
    update_query = {
        "script": {"source": source},
        "query": {
            "bool": {
                "must": [
                    {"match": {"tenantCode.keyword": {"query": tenant_code}}},
                    {
                        "range": {
                            "@timestamp": {
                                "gte": search_start_timestamp,
                                "lte": search_end_timestamp,
                            }
                        }
                    },
                ]
            }
        },
        "track_total_hits": True,
        "size": 5000,
    }

    is_updated = False  # 更新完成状态

    # 进行更新操作, 默认只需要更新一次。
    loop = 0
    while loop < 180:
        act_updated_count = demo.update_by_query(target_index, update_query)
        if int(act_updated_count) == int(exp_count):
            is_updated = True
            break

        loop += 60
        time.sleep(60)
        sync_cache_pool_upload_usages(env)

    return is_updated


if __name__ == "__main__":
    env = "SIT"
    # now = datetime.now()
    # origin_index = now.strftime('%Y.%m')
    # origin_bill_date = now - timedelta(hours=24)
    # target_index = f'billing-{origin_index}'
    # target_bill_date = origin_bill_date.strftime('%Y-%m-%d %H:%M:%S')
    # hosts = Platform_Vars.get(env)['ES']['HOSTS']
    # port = Platform_Vars.get(env)['ES']['PORT']
    # demo = ESearchHelper(hosts, port)
    # print(help(demo.es.update_by_query))
    # print(help(demo.es.update_by_query))
    # for tenant_code in ["JobWu", "JobWei"]:
    #     update_es_info_by_tenant_code(demo, target_index, tenant_code,
    #                                   target_bill_date, True, 10)
    tenant_code = "PyTestWu"
    # tax_num = '92100000JGFR16W0X5'
    # bill_date = '2022-01-01 15:30:01'
    # start_date = '2022-03-13T00:00:01'
    # end_date = '2022-01-31 23:59:59'
    # biz_type = 'verify_times'
    # channel = "api"
    # # print(
    # #     search_records_by_tenant_code(env, tenant_code, biz_type, channel,
    # #                                   start_date, end_date))
    # # update_records_by_tenant_code(env, tenant_code, bill_date )
    # print(
    #     search_records_by_tenant_code(env, tenant_code, tax_num, biz_type,
    #                                   channel))
    print(delete_records_by_tenant_code(env, tenant_code))
